package main

import (
	"context"
	"encoding/json"
	"fmt"
	"log"
	"strconv"
	"time"

	"github.com/google/uuid"
	"github.com/redis/rueidis"
)

// Timer 定时器结构
type Timer struct {
	ID        string    `json:"id"`
	Name      string    `json:"name"`
	Service   string    `json:"service"`
	Instance  string    `json:"instance"`
	Interval  int64     `json:"interval"` // 间隔时间（秒）
	NextRun   time.Time `json:"next_run"`
	Callback  string    `json:"callback"` // 回调函数名
	CreatedAt time.Time `json:"created_at"`
	Status    string    `json:"status"` // active, paused, stopped
}

// ServiceInstance 实例结构
type ServiceInstance struct {
	ID        string    `json:"id"`
	Service   string    `json:"service"`
	Host      string    `json:"host"`
	Port      int       `json:"port"`
	Status    string    `json:"status"` // active, inactive
	LastSeen  time.Time `json:"last_seen"`
	CreatedAt time.Time `json:"created_at"`
}

const (
	// service id -> instance
	serviceKeyTmpl = "DistributedTimer:%s"
	// service id, instance id -> updatetime
	serviceInstanceLockTmpl = "DistributedTimer:%s:InstanceLock:%s"
)

// TimerCallback 定时器回调函数类型
type TimerCallback func(timer *Timer) error

// DistributedTimer 分布式定时器管理器
type DistributedTimer struct {
	redisClient       rueidis.Client
	serviceInstance   *ServiceInstance
	callbacks         map[string]TimerCallback
	heartbeatStop     chan bool
	timerStop         chan bool
	failoverStop      chan bool
	ctx               context.Context
	cancel            context.CancelFunc
	LockTTL           time.Duration // 锁的TTL时间
	HeartbeatInterval time.Duration // 心跳间隔
}

// NewDistributedTimer 创建新的分布式定时器管理器
func NewDistributedTimer(redisURL, serviceID, instanceID string) (*DistributedTimer, error) {
	opt, err := rueidis.ParseURL(redisURL)
	if err != nil {
		return nil, fmt.Errorf("failed to parse Redis URL: %v", err)
	}
	rdb, err := rueidis.NewClient(opt)
	if err != nil {
		return nil, fmt.Errorf("failed to create Redis client: %v", err)
	}

	// 测试连接
	ctx := context.Background()
	err = rdb.Do(ctx, rdb.B().Ping().Build()).Error()
	if err != nil {
		return nil, fmt.Errorf("failed to connect to Redis: %v", err)
	}

	ctx, cancel := context.WithCancel(context.Background())

	serviceInstance := &ServiceInstance{
		ID:        instanceID,
		Service:   serviceID,
		Host:      "localhost", // 可以通过参数传入
		Port:      8080,        // 可以通过参数传入
		Status:    "active",
		LastSeen:  time.Now(),
		CreatedAt: time.Now(),
	}

	dt := &DistributedTimer{
		redisClient:       rdb,
		serviceInstance:   serviceInstance,
		callbacks:         make(map[string]TimerCallback),
		heartbeatStop:     make(chan bool),
		timerStop:         make(chan bool),
		failoverStop:      make(chan bool),
		ctx:               ctx,
		cancel:            cancel,
		LockTTL:           30 * time.Second, // 锁30秒过期
		HeartbeatInterval: 10 * time.Second, // 每10秒续期一次
	}

	return dt, nil
}

// RegisterCallback 注册回调函数
func (dt *DistributedTimer) RegisterCallback(name string, callback TimerCallback) {
	dt.callbacks[name] = callback
}

// Start 启动分布式定时器
func (dt *DistributedTimer) Start() error {
	// 获取实例锁
	if err := dt.acquireInstanceLock(); err != nil {
		return fmt.Errorf("failed to acquire instance lock: %v", err)
	}

	// 注册实例
	if err := dt.registerInstance(); err != nil {
		return fmt.Errorf("failed to register instance: %v", err)
	}

	// 启动心跳（锁续期）
	go dt.startHeartbeat()

	// 启动定时器检查
	go dt.startTimerChecker()

	// 启动故障转移检查
	go dt.startFailoverChecker()

	log.Printf("Distributed timer started for instance %s in service %s", dt.serviceInstance.ID, dt.serviceInstance.Service)
	return nil
}

// Stop 停止分布式定时器
func (dt *DistributedTimer) Stop() error {
	dt.cancel()

	// 非阻塞发送停止信号
	select {
	case dt.heartbeatStop <- true:
	default:
	}

	select {
	case dt.timerStop <- true:
	default:
	}

	select {
	case dt.failoverStop <- true:
	default:
	}

	// 释放实例锁
	dt.releaseInstanceLock()

	// 注销实例
	return dt.unregisterInstance()
}

// CreateTimer 创建定时器
func (dt *DistributedTimer) CreateTimer(name string, interval int64, callback string) (*Timer, error) {
	timer := &Timer{
		ID:        uuid.New().String(),
		Name:      name,
		Service:   dt.serviceInstance.Service,
		Instance:  dt.serviceInstance.ID,
		Interval:  interval,
		NextRun:   time.Now().Add(time.Duration(interval) * time.Second),
		Callback:  callback,
		CreatedAt: time.Now(),
		Status:    "active",
	}

	// 保存到Redis
	timerData, err := json.Marshal(timer)
	if err != nil {
		return nil, fmt.Errorf("failed to marshal timer: %v", err)
	}

	// 保存定时器数据
	timerKey := fmt.Sprintf("timer:%s", timer.ID)
	err = dt.redisClient.Do(dt.ctx, dt.redisClient.B().Set().Key(timerKey).Value(string(timerData)).Build()).Error()
	if err != nil {
		return nil, fmt.Errorf("failed to save timer: %v", err)
	}

	// 添加到实例的定时器列表
	instanceTimersKey := fmt.Sprintf("instance_timers:%s", dt.serviceInstance.ID)
	err = dt.redisClient.Do(dt.ctx, dt.redisClient.B().Sadd().Key(instanceTimersKey).Member(timer.ID).Build()).Error()
	if err != nil {
		return nil, fmt.Errorf("failed to add timer to instance: %v", err)
	}

	// 添加到执行队列
	scoreKey := fmt.Sprintf("timer_queue:%s", dt.serviceInstance.Service)
	err = dt.redisClient.Do(dt.ctx, dt.redisClient.B().Zadd().Key(scoreKey).ScoreMember().ScoreMember(float64(timer.NextRun.Unix()), timer.ID).Build()).Error()
	if err != nil {
		return nil, fmt.Errorf("failed to add timer to queue: %v", err)
	}

	log.Printf("Timer %s created for instance %s", timer.ID, dt.serviceInstance.ID)
	return timer, nil
}

// DeleteTimer 删除定时器
func (dt *DistributedTimer) DeleteTimer(timerID string) error {
	// 从Redis删除定时器数据
	timerKey := fmt.Sprintf("timer:%s", timerID)
	err := dt.redisClient.Do(dt.ctx, dt.redisClient.B().Del().Key(timerKey).Build()).Error()
	if err != nil {
		return fmt.Errorf("failed to delete timer: %v", err)
	}

	// 从实例定时器列表删除
	instanceTimersKey := fmt.Sprintf("instance_timers:%s", dt.serviceInstance.ID)
	err = dt.redisClient.Do(dt.ctx, dt.redisClient.B().Srem().Key(instanceTimersKey).Member(timerID).Build()).Error()
	if err != nil {
		return fmt.Errorf("failed to remove timer from instance: %v", err)
	}

	// 从执行队列删除
	scoreKey := fmt.Sprintf("timer_queue:%s", dt.serviceInstance.Service)
	err = dt.redisClient.Do(dt.ctx, dt.redisClient.B().Zrem().Key(scoreKey).Member(timerID).Build()).Error()
	if err != nil {
		return fmt.Errorf("failed to remove timer from queue: %v", err)
	}

	log.Printf("Timer %s deleted", timerID)
	return nil
}

// registerInstance 注册实例
func (dt *DistributedTimer) registerInstance() error {
	instanceData, err := json.Marshal(dt.serviceInstance)
	if err != nil {
		return fmt.Errorf("failed to marshal instance: %v", err)
	}

	// 保存实例信息
	instanceKey := fmt.Sprintf("instance:%s", dt.serviceInstance.ID)
	err = dt.redisClient.Do(dt.ctx, dt.redisClient.B().Set().Key(instanceKey).Value(string(instanceData)).Build()).Error()
	if err != nil {
		return fmt.Errorf("failed to save instance: %v", err)
	}

	// 添加到服务实例列表
	serviceInstancesKey := fmt.Sprintf(serviceKeyTmpl, dt.serviceInstance.Service)
	err = dt.redisClient.Do(dt.ctx, dt.redisClient.B().Sadd().Key(serviceInstancesKey).Member(dt.serviceInstance.ID).Build()).Error()
	if err != nil {
		return fmt.Errorf("failed to add instance to service: %v", err)
	}

	return nil
}

// unregisterInstance 注销实例
func (dt *DistributedTimer) unregisterInstance() error {
	// 删除实例信息
	instanceKey := fmt.Sprintf("instance:%s", dt.serviceInstance.ID)
	err := dt.redisClient.Do(dt.ctx, dt.redisClient.B().Del().Key(instanceKey).Build()).Error()
	if err != nil {
		return fmt.Errorf("failed to delete instance: %v", err)
	}

	// 从服务实例列表删除
	serviceInstancesKey := fmt.Sprintf(serviceKeyTmpl, dt.serviceInstance.Service)
	err = dt.redisClient.Do(dt.ctx, dt.redisClient.B().Srem().Key(serviceInstancesKey).Member(dt.serviceInstance.ID).Build()).Error()
	if err != nil {
		return fmt.Errorf("failed to remove instance from service: %v", err)
	}

	return nil
}

// startHeartbeat 启动心跳（锁续期）
func (dt *DistributedTimer) startHeartbeat() {
	ticker := time.NewTicker(dt.HeartbeatInterval)
	defer ticker.Stop()

	for {
		select {
		case <-dt.heartbeatStop:
			return
		case <-dt.ctx.Done():
			return
		case <-ticker.C:
			// 续期实例锁
			if err := dt.renewInstanceLock(); err != nil {
				log.Printf("Failed to renew instance lock: %v", err)
				// 尝试重新获取锁
				if err := dt.acquireInstanceLock(); err != nil {
					log.Printf("Failed to reacquire instance lock: %v", err)
				}
			}
			// 更新实例信息
			dt.serviceInstance.LastSeen = time.Now()
			instanceData, _ := json.Marshal(dt.serviceInstance)
			instanceKey := fmt.Sprintf("instance:%s", dt.serviceInstance.ID)
			dt.redisClient.Do(dt.ctx, dt.redisClient.B().Set().Key(instanceKey).Value(string(instanceData)).Build())
		}
	}
}

// startTimerChecker 启动定时器检查
func (dt *DistributedTimer) startTimerChecker() {
	ticker := time.NewTicker(1 * time.Second)
	defer ticker.Stop()

	for {
		select {
		case <-dt.timerStop:
			return
		case <-dt.ctx.Done():
			return
		case <-ticker.C:
			dt.checkAndExecuteTimers()
		}
	}
}

// checkAndExecuteTimers 检查并执行定时器
func (dt *DistributedTimer) checkAndExecuteTimers() {
	now := time.Now().Unix()
	scoreKey := fmt.Sprintf("timer_queue:%s", dt.serviceInstance.Service)

	// 获取需要执行的定时器
	result, err := dt.redisClient.Do(dt.ctx, dt.redisClient.B().Zrangebyscore().Key(scoreKey).Min("0").Max(strconv.FormatInt(now, 10)).Build()).AsStrSlice()
	if err != nil {
		log.Printf("Failed to get timers to execute: %v", err)
		return
	}

	for _, timerID := range result {
		dt.executeTimer(timerID)
	}
}

// executeTimer 执行定时器
func (dt *DistributedTimer) executeTimer(timerID string) {
	// 获取定时器信息
	timerKey := fmt.Sprintf("timer:%s", timerID)
	timerData, err := dt.redisClient.Do(dt.ctx, dt.redisClient.B().Get().Key(timerKey).Build()).ToString()
	if err != nil {
		if !rueidis.IsRedisNil(err) {
			log.Printf("Failed to get timer %s: %v", timerID, err)
		}
		return
	}

	var timer Timer
	err = json.Unmarshal([]byte(timerData), &timer)
	if err != nil {
		log.Printf("Failed to unmarshal timer %s: %v", timerID, err)
		return
	}

	// 检查定时器状态
	if timer.Status != "active" {
		return
	}

	// 执行回调函数
	if callback, exists := dt.callbacks[timer.Callback]; exists {
		go func() {
			if err := callback(&timer); err != nil {
				log.Printf("Timer %s callback failed: %v", timerID, err)
			}
		}()
	} else {
		log.Printf("Callback %s not found for timer %s", timer.Callback, timerID)
	}

	// 更新下次执行时间
	timer.NextRun = time.Now().Add(time.Duration(timer.Interval) * time.Second)
	updatedTimerData, _ := json.Marshal(timer)
	dt.redisClient.Do(dt.ctx, dt.redisClient.B().Set().Key(timerKey).Value(string(updatedTimerData)).Build())

	// 更新执行队列
	scoreKey := fmt.Sprintf("timer_queue:%s", dt.serviceInstance.Service)
	dt.redisClient.Do(dt.ctx, dt.redisClient.B().Zadd().Key(scoreKey).ScoreMember().ScoreMember(float64(timer.NextRun.Unix()), timerID).Build())

	log.Printf("Timer %s executed, next run at %v", timerID, timer.NextRun)
}

// startFailoverChecker 启动故障转移检查（基于锁检测）
func (dt *DistributedTimer) startFailoverChecker() {
	ticker := time.NewTicker(10 * time.Second)
	defer ticker.Stop()

	for {
		select {
		case <-dt.failoverStop:
			return
		case <-dt.ctx.Done():
			return
		case <-ticker.C:
			dt.checkFailedInstancesByLock()
		}
	}
}

// acquireInstanceLock 获取实例锁
func (dt *DistributedTimer) acquireInstanceLock() error {
	lockKey := fmt.Sprintf(serviceInstanceLockTmpl, dt.serviceInstance.Service, dt.serviceInstance.ID)
	result, err := dt.redisClient.Do(dt.ctx, dt.redisClient.B().Set().Key(lockKey).Value(dt.serviceInstance.ID).Nx().Ex(dt.LockTTL).Build()).AsBool()
	if err != nil {
		return fmt.Errorf("failed to acquire lock: %v", err)
	}
	if !result {
		return fmt.Errorf("instance lock already exists")
	}
	return nil
}

// renewInstanceLock 续期实例锁
func (dt *DistributedTimer) renewInstanceLock() error {
	lockKey := fmt.Sprintf(serviceInstanceLockTmpl, dt.serviceInstance.Service, dt.serviceInstance.ID)
	result, err := dt.redisClient.Do(dt.ctx, dt.redisClient.B().Expire().Key(lockKey).Seconds(int64(dt.LockTTL.Seconds())).Build()).AsBool()
	if err != nil {
		return fmt.Errorf("failed to renew lock: %v", err)
	}
	if !result {
		return fmt.Errorf("lock does not exist")
	}
	return nil
}

// releaseInstanceLock 释放实例锁
func (dt *DistributedTimer) releaseInstanceLock() error {
	lockKey := fmt.Sprintf(serviceInstanceLockTmpl, dt.serviceInstance.Service, dt.serviceInstance.ID)
	err := dt.redisClient.Do(dt.ctx, dt.redisClient.B().Del().Key(lockKey).Build()).Error()
	if err != nil {
		return fmt.Errorf("failed to release lock: %v", err)
	}
	return nil
}

// checkFailedInstancesByLock 通过锁检测故障实例
func (dt *DistributedTimer) checkFailedInstancesByLock() {
	serviceInstancesKey := fmt.Sprintf(serviceKeyTmpl, dt.serviceInstance.Service)
	instanceIDs, err := dt.redisClient.Do(dt.ctx, dt.redisClient.B().Smembers().Key(serviceInstancesKey).Build()).AsStrSlice()
	if err != nil {
		log.Printf("Failed to get service instances: %v", err)
		return
	}

	for _, instanceID := range instanceIDs {
		if instanceID == dt.serviceInstance.ID {
			continue
		}

		// 尝试获取其他实例的锁来检测是否故障
		lockKey := fmt.Sprintf(serviceInstanceLockTmpl, dt.serviceInstance.Service, instanceID)
		result, err := dt.redisClient.Do(dt.ctx, dt.redisClient.B().Set().Key(lockKey).Value(dt.serviceInstance.ID).Nx().Ex(dt.LockTTL).Build()).AsBool()
		if err != nil {
			continue
		}
		if result {
			// 成功获取锁，说明原实例已故障
			log.Printf("Detected failed instance: %s", instanceID)
			dt.handleFailedInstance(instanceID)
			// 释放临时锁
			dt.redisClient.Do(dt.ctx, dt.redisClient.B().Del().Key(lockKey).Build())
		}
	}
}

// handleFailedInstance 处理故障实例
func (dt *DistributedTimer) handleFailedInstance(failedInstanceID string) {
	log.Printf("Detected failed instance: %s, starting failover", failedInstanceID)

	// 获取故障实例的定时器
	instanceTimersKey := fmt.Sprintf("instance_timers:%s", failedInstanceID)
	timerIDs, err := dt.redisClient.Do(dt.ctx, dt.redisClient.B().Smembers().Key(instanceTimersKey).Build()).AsStrSlice()
	if err != nil {
		log.Printf("Failed to get timers for failed instance %s: %v", failedInstanceID, err)
		return
	}

	// 转移定时器到当前实例
	for _, timerID := range timerIDs {
		dt.transferTimer(timerID, failedInstanceID)
	}

	// 清理故障实例
	dt.cleanupFailedInstance(failedInstanceID)
}

// transferTimer 转移定时器
func (dt *DistributedTimer) transferTimer(timerID, failedInstanceID string) {
	// 获取定时器信息
	timerKey := fmt.Sprintf("timer:%s", timerID)
	timerData, err := dt.redisClient.Do(dt.ctx, dt.redisClient.B().Get().Key(timerKey).Build()).ToString()
	if err != nil {
		log.Printf("Failed to get timer %s for transfer: %v", timerID, err)
		return
	}

	var timer Timer
	err = json.Unmarshal([]byte(timerData), &timer)
	if err != nil {
		log.Printf("Failed to unmarshal timer %s for transfer: %v", timerID, err)
		return
	}

	// 更新定时器的实例ID
	timer.Instance = dt.serviceInstance.ID
	updatedTimerData, _ := json.Marshal(timer)
	dt.redisClient.Do(dt.ctx, dt.redisClient.B().Set().Key(timerKey).Value(string(updatedTimerData)).Build())

	// 从故障实例的定时器列表删除
	failedInstanceTimersKey := fmt.Sprintf("instance_timers:%s", failedInstanceID)
	dt.redisClient.Do(dt.ctx, dt.redisClient.B().Srem().Key(failedInstanceTimersKey).Member(timerID).Build())

	// 添加到当前实例的定时器列表
	currentInstanceTimersKey := fmt.Sprintf("instance_timers:%s", dt.serviceInstance.ID)
	dt.redisClient.Do(dt.ctx, dt.redisClient.B().Sadd().Key(currentInstanceTimersKey).Member(timerID).Build())

	log.Printf("Timer %s transferred from instance %s to %s", timerID, failedInstanceID, dt.serviceInstance.ID)
}

// cleanupFailedInstance 清理故障实例
func (dt *DistributedTimer) cleanupFailedInstance(failedInstanceID string) {
	// 删除实例信息
	instanceKey := fmt.Sprintf("instance:%s", failedInstanceID)
	dt.redisClient.Do(dt.ctx, dt.redisClient.B().Del().Key(instanceKey).Build())

	// 从服务实例列表删除
	serviceInstancesKey := fmt.Sprintf(serviceKeyTmpl, dt.serviceInstance.Service)
	dt.redisClient.Do(dt.ctx, dt.redisClient.B().Srem().Key(serviceInstancesKey).Member(failedInstanceID).Build())

	// 删除实例定时器列表
	instanceTimersKey := fmt.Sprintf("instance_timers:%s", failedInstanceID)
	dt.redisClient.Do(dt.ctx, dt.redisClient.B().Del().Key(instanceTimersKey).Build())

	log.Printf("Failed instance %s cleaned up", failedInstanceID)
}

// GetInstanceTimers 获取实例的所有定时器
func (dt *DistributedTimer) GetInstanceTimers() ([]*Timer, error) {
	instanceTimersKey := fmt.Sprintf("instance_timers:%s", dt.serviceInstance.ID)
	timerIDs, err := dt.redisClient.Do(dt.ctx, dt.redisClient.B().Smembers().Key(instanceTimersKey).Build()).AsStrSlice()
	if err != nil {
		return nil, fmt.Errorf("failed to get timer IDs: %v", err)
	}

	var timers []*Timer
	for _, timerID := range timerIDs {
		timerKey := fmt.Sprintf("timer:%s", timerID)
		timerData, err := dt.redisClient.Do(dt.ctx, dt.redisClient.B().Get().Key(timerKey).Build()).ToString()
		if err != nil {
			continue
		}

		var timer Timer
		err = json.Unmarshal([]byte(timerData), &timer)
		if err != nil {
			continue
		}

		timers = append(timers, &timer)
	}

	return timers, nil
}

// GetServiceInstances 获取服务的所有实例
func (dt *DistributedTimer) GetServiceInstances() ([]*ServiceInstance, error) {
	serviceInstancesKey := fmt.Sprintf(serviceKeyTmpl, dt.serviceInstance.Service)
	instanceIDs, err := dt.redisClient.Do(dt.ctx, dt.redisClient.B().Smembers().Key(serviceInstancesKey).Build()).AsStrSlice()
	if err != nil {
		return nil, fmt.Errorf("failed to get instance IDs: %v", err)
	}

	var serviceInstances []*ServiceInstance
	for _, instanceID := range instanceIDs {
		instanceKey := fmt.Sprintf("instance:%s", instanceID)
		instanceData, err := dt.redisClient.Do(dt.ctx, dt.redisClient.B().Get().Key(instanceKey).Build()).ToString()
		if err != nil {
			continue
		}

		var serviceInstance ServiceInstance
		err = json.Unmarshal([]byte(instanceData), &serviceInstance)
		if err != nil {
			continue
		}

		serviceInstances = append(serviceInstances, &serviceInstance)
	}

	return serviceInstances, nil
}
